
import util.Util._
import java.text.SimpleDateFormat
import java.util.{Calendar, TimeZone}
import java.text.MessageFormat.format
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.hive._
import org.json4s._
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._
import  org.json4s.JsonAST._
object Convent2Json {
  def main(args: Array[String]) {

    // Validate args
    if (args.length < 5) {
      println("Usage: [2015-05-25|date] [tablename] [fieldname] [savehdfsfile] [daily|all |option]")
      sys.exit(1)
    }
    val date_str = args(0).toString
    val tableName = args(1).toString
    val fieldname = args(2).toString.replace("#","  ")
    val filePath = getConfig("jsonsql.hdfsdir") + args(3).toString
    val option = args(4).toString

    val date_sdf = new SimpleDateFormat("yyyy-MM-dd")
    val daily_date = date_sdf.parse(date_str)
    val calender = Calendar.getInstance(TimeZone.getTimeZone("GMT+8"))
    calender.setTime(daily_date)
    val year = calender.get(Calendar.YEAR)
    val month = calender.get(Calendar.MONTH) + 1
    val day = calender.get(Calendar.DAY_OF_MONTH)

    var sql_str = ""
    if (option == "all") {
      sql_str = format(getConfig("jsonsql.day_total"), tableName, fieldname, date_str)
    }
    else {
      sql_str = format(getConfig("jsonsql.day_list"), tableName, fieldname, year.toString, month.toString, day.toString, date_str)
    }
    val sparkConf = new SparkConf().setAppName("convent2json")
    val sc = new SparkContext(sparkConf)

    val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
    import sqlContext._
    if (!sql_str.isEmpty) {
      val hdfs = FileSystem.get(new Configuration())
      if (hdfs.exists(new Path(filePath))) hdfs.delete(new Path(filePath), true)
      sqlContext.sql(sql_str).map(d=>d.schema.fields)
      sqlContext.sql(sql_str).toJSON.coalesce(1, shuffle = true).saveAsTextFile(filePath)
    }

  }

}
